package com.atbeijing.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;


/**
 * 从kafka读取ods层用户行为日志数据-->dwd层 kafka三个主题
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 创建流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度,并行度与kafka分区数相同即可
        env.setParallelism(4);

//        //TODO 2.设置检查点
//        //2.1设置flink精准一次性保证(默认) 每5000ms开始一次checkpoint
//        env.enableCheckpointing(5*1000L, CheckpointingMode.EXACTLY_ONCE);
//        //2.2checkpoint必须在1分钟内完成,否则被抛弃
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
//        //2.3设置取消job后是否保留检查点,这里是保存
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        //2.4设置检查点重启策略,最多重启三次,间隔3秒
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
//        //2.5设置检查点存放位置-->状态后端  内存|文件系统|rockDB
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop202:8020/gmall/ck"));
//        //2.6hadoop 用户
//        System.setProperty("HADOOP_USER_NAME","wjg");

        //TODO 3.从Kafka中读取数据,将数据转为json对象
        //3.1指定消费者配置信息
        String groupId = "ods_dwd_base_log_app";
        String topic = "ods_base_log";

        //3.2flink从kafka读取数据
        FlinkKafkaConsumer kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);

        //3.3 将消费到的数据封装到流中
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //3.4将kafka消息转为json对象
        //3.4.1 匿名内部类
//        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaDS.map(new MapFunction<String, JSONObject>() {
//            @Override
//            public JSONObject map(String value) throws Exception {
//                return JSON.parseObject(value);
//            }
//        });
        //3.4.2 lambda表达式
//        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaDS.map(jsonStr -> JSON.parseObject(jsonStr));

        //3.4.3 方法默认调用
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaDS.map(JSON::parseObject);

        //TODO 4.新老访客状态修复
        //思路：将设备上次访问时间记录到状态中，下次再访问的时候，从状态中获取时间，如果有说明访问过，修复；
        // 如果没有，说明没有访问过，将这次访问的时间记录到状态中

        //4.1按照mid分组
        KeyedStream<JSONObject, String> keyedDS = jsonObj.keyBy(r -> r.getJSONObject("common").getString("mid"));

        //4.2状态修复
        SingleOutputStreamOperator<JSONObject> jsonObjWithFlagDS = keyedDS.map(new RichMapFunction<JSONObject, JSONObject>() {
            //存放第一次访问日期,键控状态,每隔key共享一个状态,这里同一个mid共享一个状态
            private ValueState<String> lastVisitDateState;
            //时间戳格式化
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDateState", Types.STRING));
                sdf = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {
                //业务日志的日期,当前日期
                String curDate = sdf.format(value.getLong("ts"));
                //是否时第一次访问
                String isNew = value.getJSONObject("common").getString("is_new");
                //日志里"is_new": "1",标记的是新访客，那么有可能状态是不准确的，需要修复
                if (isNew.equals("1")) {
                    if (lastVisitDateState.value() == null) {
                        //第一次访问
                        lastVisitDateState.update(curDate);
                    } else {
                        //不是第一次访问,如果日期不是今天,修改日志json数据
                        if (!lastVisitDateState.equals(curDate)){
                            value.getJSONObject("common").put("is_new", "0");
                        }
                    }
                }
                return value;
            }
        });

        //TODO 5.分流   启动---启动侧输出流   曝光---曝光侧输出流   页面---主流
        //定义侧输出流
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag"){};

        SingleOutputStreamOperator<String> splitDS = jsonObjWithFlagDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                String jsonStr = value.toJSONString();
                //如果不是启动日志，都属于页面日志类型,页面日志中包含曝光日志
                JSONObject start = value.getJSONObject("start");
                if (start != null && start.size() > 0) {
                    //启动日志输出到侧输出流
                    ctx.output(startTag, jsonStr);
                } else {
                    //页面日志输出到主流
                    out.collect(jsonStr);

                    //获取取页面id
                    String pageId = value.getJSONObject("page").getString("page_id");
                    //判断是否是曝光日志
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        //为曝光页面添加页面id
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject jsonObject = displays.getJSONObject(i);
                            jsonObject.put("page_id", pageId);
                            //将曝光日志输出到侧输出流
                            ctx.output(displayTag, jsonObject.toJSONString());
                        }
                    }
                }
            }
        });

        //TODO 6. 将不同流的数据写到Kafka的dwd的不同主题中
        //声明kafka主题
        String startSinkTopic = "dwd_start_log";
        String displaySinkTopic = "dwd_display_log";
        String pageSinkTopic = "dwd_page_log";

        //获取侧输出流
        DataStream<String> startSideOutput = splitDS.getSideOutput(startTag);
        DataStream<String> displaySideOutput = splitDS.getSideOutput(displayTag);

        //向kafka写数据
        startSideOutput.print("启动日志流>>>");
        displaySideOutput.print("曝光日志流>>>");
        splitDS.print("页面主流>>>");

        startSideOutput.addSink(MyKafkaUtil.getKafkaSink(startSinkTopic));
        displaySideOutput.addSink(MyKafkaUtil.getKafkaSink(displaySinkTopic));
        splitDS.addSink(MyKafkaUtil.getKafkaSink(pageSinkTopic));

        env.execute();
    }

}
